package com.heima.Service.impl;

import com.alibaba.fastjson.JSON;
import com.aliyuncs.utils.StringUtils;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.heima.Common.Lock;
import com.heima.Mapper.TaskinfoMapper;
import com.heima.Mapper.TaskinfologsMapper;
import com.heima.Service.ScheduleSerice;
import com.heima.common.exception.ScheduleConstants;
import com.heima.model.schedule.pojos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.StringRedisConnection;
import org.springframework.data.redis.core.*;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.util.*;

@Service
@Slf4j
@Transactional
public class ScheduleSerivce implements ScheduleSerice {
    @Autowired
    StringRedisTemplate redisTemplate;
    @Autowired
    TaskinfoMapper taskinfoMapper;

    @Autowired
    TaskinfologsMapper taskinfologsMapper;

    @Override
    public boolean AddTask(Task task) {
        boolean flag = false;
        long timeMillis = System.currentTimeMillis();
//        添加任务到数据库
        Long taskid = AddTaskToMysql(task);
//        添加任务到redis
        boolean flag2 = AddTAskToCahche(task,timeMillis);
        flag = (taskid != null) && flag2;
        return flag;
    }


    @Override
    public boolean DeleteTask(Long id) {
        boolean flag = false;
        try {
//            删除sql数据
            Task task = UpdateOrDeleteInSql(id, ScheduleConstants.CANCELLED);
//            删除redis数据
            DeleteINChache(task);
            flag = true;
        } catch (Exception e) {
            log.error("删除任务失败");
        }
        return flag;
    }


    @Override
    public Task ExecuteTask(Integer taskType,Integer priority) {
        try {
//        消费任务
//        从redis的list里提取数据并删除
            String key = taskType + "_" + priority;
//            cacheService.lRightPop()
            String rightPop = redisTemplate.opsForList().rightPop(ScheduleConstants.TOPIC+key);
            if(StringUtils.isNotEmpty(rightPop))
            {
                Task parseObject = JSON.parseObject(rightPop, Task.class);
//        修改TaskinfoLogs状态并删除sql
                UpdateOrDeleteInSql(parseObject.getTaskId(), ScheduleConstants.EXECUTED);
                return parseObject;
            }

        } catch (Exception e) {
            log.error("任务消费失败");
        }
        return null;
    }

//    定时同步未来的数据
    @Override
    @Scheduled(cron = "0 */1 * * * ?")//每分钟执行一次
    public boolean timingPush() {
//        String token = Lock.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
//        if(StringUtils.isNotEmpty(token)) {
            long timeMillis = System.currentTimeMillis();
            try {
                //        查询所有未来数据
                String parrten = ScheduleConstants.FUTURE + "*";
                Set<String> keys = SelectTask(parrten);
//        选出过期数据
                for (String key : keys) {
                    String topickey = ScheduleConstants.TOPIC + key.split(ScheduleConstants.FUTURE)[1];
                    Set<String> range = redisTemplate.opsForZSet().rangeByScore(key, 0, timeMillis);
                    if (range != null)
                        datasynchronize(key, topickey, range);
                }
                log.info("redis内部数据同步成功");
                return true;
            } catch (Exception e) {
                log.error("redis内部数据同步失败");
            }
//        }
        return false;
    }

//    同步数据库和redis的数据
    @Override
    @Scheduled(cron = "0 */5 * * * ?")//每5分钟执行一次
    public boolean timingPushSQLAndChache() {
        Calendar calendar=Calendar.getInstance();
        long timeMillis = System.currentTimeMillis();
//        删除全部redis的数据（没执行的也不要紧，sql里还会重新同步）
        String futureparrten=ScheduleConstants.FUTURE+"*";
        String topicparrten=ScheduleConstants.TOPIC+"*";
        ClearRedis(futureparrten,topicparrten);
//        查询符合条件的数据
        QueryWrapper<Taskinfo> wrapper=new QueryWrapper<>();
        wrapper.lambda().lt(Taskinfo::getExecuteTime,calendar.getTime());
        List<Taskinfo> taskinfos = taskinfoMapper.selectList(wrapper);
//        数据同步
        datasynchronizeInSQLAndChache(taskinfos,timeMillis);
        log.info("同步数据库和redis的数据成功");
        return false;
    }

    //    查询所有未来数据数据
    private Set<String> SelectTask(String parrten) {
        Set<String> keys = redisTemplate.execute((RedisCallback<Set<String>>) connection -> {

            Set<String> result = new HashSet<>();
            try (Cursor<byte[]> cursor = connection.scan(new ScanOptions.ScanOptionsBuilder()
                    .match(parrten).count(10000).build())) {
                while (cursor.hasNext()) {
                    result.add(new String(cursor.next()));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return result;

        });
        return  keys;
    }


    //        添加任务到redis
    private boolean AddTAskToCahche(Task task,long timeMillis ) {
        boolean flag = false;
        String key = task.getTaskType() + "_" + task.getPriority();
//        任务时间是否小于等于当前时间
        if (task.getExecuteTime() <= timeMillis) {
            ListOperations listOperations = redisTemplate.opsForList();
            listOperations.leftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
        }
//        任务时间是否小于等于预设时间(5分钟后)
        else if (task.getExecuteTime() <= timeMillis + 3000000) {
            ZSetOperations zSetOperations = redisTemplate.opsForZSet();
            zSetOperations.add(ScheduleConstants.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
        }
        return flag;
    }


    //        添加任务到数据库
    private Long AddTaskToMysql(Task task) {
        boolean flag = false;
//        添加任务到任务表
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        taskinfoMapper.insert(taskinfo);
        task.setTaskId(taskinfo.getTaskId());
//        添加任务日志
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo, taskinfoLogs);
        taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
        taskinfoLogs.setVersion(1);
        taskinfologsMapper.insert(taskinfoLogs);
        return task.getTaskId();
    }


    //            删除sql数据
    private Task UpdateOrDeleteInSql(Long id, Integer status) {
        boolean flag = false;
//            删除sql数据
        taskinfoMapper.deleteById(id);
        TaskinfoLogs taskinfoLogs = taskinfologsMapper.selectById(id);
        taskinfoLogs.setStatus(status);
        taskinfologsMapper.updateById(taskinfoLogs);
        Task task = new Task();
        BeanUtils.copyProperties(taskinfoLogs, task);
        return task;
    }


    //            删除redis数据
    private boolean DeleteINChache(Task task) {
        boolean flag = false;
        String key = task.getTaskType() + "_" + task.getPriority();
        Calendar calendar = Calendar.getInstance();
//        任务时间是否小于等于当前时间
        if (task.getExecuteTime() <= calendar.getTimeInMillis()) {
            ListOperations listOperations = redisTemplate.opsForList();
            listOperations.remove(ScheduleConstants.TOPIC + key, 0, JSON.toJSONString(task));
        }
//        任务时间是否小于等于预设时间(5分钟后)
        else if (task.getExecuteTime() <= calendar.getTimeInMillis() + 3000000) {
            ZSetOperations zSetOperations = redisTemplate.opsForZSet();
            zSetOperations.remove(ScheduleConstants.FUTURE + key, JSON.toJSONString(task));
        }
        return flag;
    }

//    redis内部数据同步
    private List<Object> datasynchronize(String futurekey,String tocipkey,Set<String> tasks)
    {
        List<Object> objects = redisTemplate.executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
                StringRedisConnection stringRedisConnection = (StringRedisConnection)redisConnection;
                for (String task:tasks)
                {
                    stringRedisConnection.rPush(tocipkey,task);
                    stringRedisConnection.zRem(futurekey,task);
                }
                return null;
            }
        });
        return objects;
    }


//    清空redis
    private boolean ClearRedis(String futureparrten ,String topicparrten )
    {
        Set<String> futuretasks = SelectTask(futureparrten);
        Set<String> topictasks = SelectTask(topicparrten);
        List<Object> objects = redisTemplate.executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
                StringRedisConnection stringRedisConnection = (StringRedisConnection) redisConnection;
                for (String futuretask:futuretasks)
                {
                    stringRedisConnection.del(futuretask);
                }
                for (String topictask:topictasks)
                {
                    stringRedisConnection.del(topictask);
                }
                return null;
            }
        });
        return true;
    }

//    数据库和redis之间的数据同步
    private boolean datasynchronizeInSQLAndChache(List<Taskinfo> list,long timeMillis)
    {
        try{
            for(Taskinfo taskinfo:list)
            {
                Task task=new Task();
                BeanUtils.copyProperties(taskinfo,task);
                task.setExecuteTime(taskinfo.getExecuteTime().getTime());
                AddTAskToCahche(task,timeMillis);
            }
            return true;
        }
        catch (Exception e)
        {
            log.error("SQl与redis数据同步失败");
        }
        return false;
    }
}
